//
//  Custom routing Router to Dealer (ROUTER to DEALER)
//
//  While this example runs in a single process, that is just to make
//  it easier to start and stop the example. Each thread has its own
//  context and conceptually acts as a separate process.
//

package main

import (
	"fmt"
	"os"
	"zmq"
	"rand"
	"time"
)

func worker_task(id string) os.Error {
	context, err := zmq.Init(1)
	if err != nil {
		return err
	}
	defer context.Close()

	worker, err := context.Socket(zmq.DEALER)
	if err != nil {
		return err
	}
	defer worker.Close()

	err = worker.SetOptionVal(zmq.IDENTITY, id)
	if err != nil {
		return err
	}
	err = worker.Connect("tcp://localhost:5557")
	if err != nil {
		return err
	}
	for i := 0; ; i++ {
		data, err := worker.RecvString(0)
		if err != nil {
			return err
		}
		if data == "END" {
			fmt.Printf("%s received: %d\n", id, i)
			break
		}
	}
	return nil
}

func zmq_main() os.Error {
	context, err := zmq.Init(1)
	if err != nil {
		return err
	}
	defer context.Close()

	client, err := context.Socket(zmq.ROUTER)
	if err != nil {
		return err
	}
	defer client.Close()

	err = client.Bind("tcp://*:5557")
	if err != nil {
		return err
	}
	exit := make(chan int, 2)
	go func() {
		err := worker_task("A")
		if err != nil {
			errch <- err
		} else {
			exit <- 0
		}
	}()
	go func() {
		err := worker_task("B")
		if err != nil {
			errch <- err
		} else {
			exit <- 1
		}
	}()

	zmq.Sleep(1e9)

	r := rand.New(rand.NewSource(time.Nanoseconds()))
	na,nb := 0,0
	var id string
	for i := 0; i < 10; i++ {
		if r.Intn(3) > 0 {
			na++
			id = "A"
		} else {
			nb++
			id = "B"
		}
		err = client.SendAll([]byte(id),[]byte("This is the workload"))
		//err = client.SendAllString(id,"This is the workload")
		if err != nil {
			return err
		}
	}
	err = client.SendAllString("A","END")
	if err != nil {
		return err
	}
	err = client.SendAllString("B","END")
	if err != nil {
		return err
	}
	fmt.Printf("send a %d, send b %d\n",na,nb)
	<-exit
	return nil
}

var errch = make(chan os.Error)

func main() {
	fmt.Println("Durable subscriber")
	exit := make(chan bool)
	go func() {
		for {
			err := <-errch
			if err != nil {
				fmt.Println(err)
				os.Exit(1)
			}
		}
	}()
	go func() {
		err := zmq_main()
		if err != nil {
			errch <- err
		} else {
			exit <- true
		}
	}()
	<-exit
}
